package com.sjtu.service;

import com.sjtu.constants.ApiConstants;
import com.sjtu.constants.TaskConstants;
import com.sjtu.dao.model.JobRunHistory;
import com.sjtu.dao.model.TaskInfo;
import com.sjtu.dao.repository.JobRunHistoryRepository;
import com.sjtu.form.JobInvokeForm;
import com.sjtu.util.JpaUtil;
import com.sjtu.zookeeper.InstanceDetails;
import com.sjtu.zookeeper.ServiceDiscoverUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.ServiceProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.context.ApplicationContext;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * nerine
 * 1/29/2020
 */
@Slf4j
@Service
public class TaskSendService {
    @Autowired
    private DiscoveryClient discoveryClient;
    @Autowired
    private WebClient webClient;
    @Autowired
    private ApplicationContext context;
    @Autowired
    private CuratorFramework curatorFramework;
    @Autowired
    private JobRunHistoryRepository jobRunHistoryRepository;

    public static final String ZK_PATH = "/job-lock";

    /**
     * sendTask
     * @param taskInfo
     */
    public void sendTask(TaskInfo taskInfo) throws Exception {
        String lockPath = lockPath(taskInfo);
        InterProcessMutex lock = new InterProcessMutex(curatorFramework, lockPath);
        if(!lock.acquire(taskInfo.getLeastInterval(), TimeUnit.SECONDS)) {
            log.info("lockpath {} not get lock", lockPath);
        }
        log.info("lockpath {} get lock", lockPath);

        JobRunHistory jobRunHistory = new JobRunHistory();
        jobRunHistory.setTaskInfo(taskInfo);
        jobRunHistory.setClientInfo(taskInfo.getClientInfo());
        jobRunHistory.setStatus(TaskConstants.TaskRunStatus.NOT_RUN);
        JpaUtil.setModifyInfo("system", jobRunHistory);
        try {
            ServiceProvider<InstanceDetails> provider =
                    ServiceDiscoverUtil.getServiceProvider(taskInfo.getClientInfo().getSystemId());


            if(taskInfo.getTaskType() == TaskConstants.TaskType.SINGLE) {
                final String sequenceId = UUID.randomUUID().toString();

                JobInvokeForm form = new JobInvokeForm(sequenceId,
                        taskInfo.getClientInfo().getSystemId(),
                        taskInfo.getBeanName());
                Mono<JobInvokeForm> mono = Mono.just(form);

                jobRunHistory.setCommonRunId(sequenceId);
                jobRunHistory.setIdJobRunHistory(sequenceId);

                send(provider.getInstance(), mono, jobRunHistory);
            }else {
                jobRunHistory.setCommonRunId(UUID.randomUUID().toString());

                provider.getAllInstances().forEach(serviceInstance -> {
                    final String sequenceId = UUID.randomUUID().toString();
                    JobInvokeForm form = new JobInvokeForm(sequenceId,
                            taskInfo.getClientInfo().getSystemId(),
                            taskInfo.getBeanName());
                    Mono<JobInvokeForm> mono = Mono.just(form);

                    jobRunHistory.setIdJobRunHistory(sequenceId);
                    send(serviceInstance, mono, jobRunHistory);
                });
            }
        }catch (Exception e) {
            log.error("error in get service instance: ", e);
            jobRunHistory.setStatus(TaskConstants.TaskRunStatus.NOT_RUN);
            jobRunHistory.setRemark(ExceptionUtils.getRootCauseMessage(e));
            jobRunHistoryRepository.save(jobRunHistory);
        } finally {
            lock.release();
            log.info("lockpath {} release lock", lockPath);
        }
    }

    String lockPath(TaskInfo taskInfo) {
        final String split = "-";
        return ZK_PATH + "/" + taskInfo.getClientInfo().getSystemId() + split +
                taskInfo.getBeanName();
    }

    void send(ServiceInstance<InstanceDetails> serviceInstance, Mono<JobInvokeForm> mono,
                    JobRunHistory jobRunHistory) {
        try {
            jobRunHistory.setInstance(serviceInstance.getAddress() + ":" + serviceInstance.getPort());
            //先保存一下防止job很快运行完成后回推时这里的事务还没提交
            jobRunHistoryRepository.save(jobRunHistory);

            short result = webClient.post()
                    .uri(concatHttpUrl(serviceInstance))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(mono, JobInvokeForm.class)
                    .retrieve()
                    .bodyToMono(Short.class)
                    .block().shortValue();
            if(result == ApiConstants.SUCCESS) {
                jobRunHistory.setStatus(TaskConstants.TaskRunStatus.RUNNING);
            }else if(result == ApiConstants.FAIL || result == ApiConstants.NO_SERVICE_FOUND) {
                jobRunHistory.setStatus(TaskConstants.TaskRunStatus.FAIL);
                jobRunHistory.setRemark("client return code: " + result);
            }
        }catch (Exception e) {
            log.error("error in task send:", e);
            jobRunHistory.setStatus(TaskConstants.TaskRunStatus.FAIL);
            jobRunHistory.setRemark(ExceptionUtils.getRootCauseMessage(e));
        }
        jobRunHistoryRepository.save(jobRunHistory);
    }

    String concatHttpUrl(ServiceInstance<InstanceDetails> serviceInstance) {
        return "http://" + serviceInstance.getAddress() + ":" + serviceInstance.getPort() + ApiConstants.CLIENT_INVOKE_URL;
    }
}
